/**
 * Copyright (C) 2010 Neofonie GmbH
 *
 * This programm is free software; you can redistribute it and/or modify
 * it under the terms of the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package eu.dicodeproject.twitterstream.source;

import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_DEFAULT_ERROR_THRESHOLD;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_DEFAULT_QUEUE_SIZE;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_DEFAULT_RECEIVE_TIMEOUT;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_DEFAULT_TAKE_TIMEOUT;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_ERROR_THRESHOLD;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_QUEUE_SIZE;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_RECEIVE_TIMEOUT;
import static eu.dicodeproject.twitterstream.util.Configuration.TWEET_TAKE_TIMEOUT;

import java.util.Iterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;
import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.conf.ConfigurationBuilder;
import eu.dicodeproject.twitterstream.Parameters;
import eu.dicodeproject.twitterstream.util.Configuration;

/**
 * Contacts the twitter sampling streaming api for iterative tweet generation. Username and password must
 * be provided in the properties, all else is handled locally. Semantics for sampling: There is just
 * one single iterator talking to twitter. As a result concurrently contacting this iterator results in
 * each thread getting only a subset of all available tweets.
 */
public class TweetSamplingStream implements TweetSource, StatusListener {
  /**Logger for this class.*/
  private static final Logger LOG = LoggerFactory.getLogger(TweetSamplingStream.class);
  /**Pointer to our single iterator.*/
  private final TweetSamplingStreamIterator iterator = new TweetSamplingStreamIterator();
  /**Username to use when contacting twitter.*/
  private final String username;
  /**Password to use when contacting twitter.*/
  private final String password;
  /**Queue holding tweets we already retrieved.*/
  private final BlockingQueue<Status> tweetQueue;
  /**Set to true if retrieval fails or connection breaks down.*/
  private final AtomicInteger failures;
  /**Link to configuration provider.*/
  private final Configuration configuration;

  /**
   * Init with correct username and password.
   * */
  public TweetSamplingStream(final Parameters parameters) {
    this.username = parameters.getUsername();
    this.password = parameters.getPassword();
    this.failures = new AtomicInteger(0);
    this.configuration = Configuration.getInstance();

    int queueSize = this.configuration.getIntValue(TWEET_QUEUE_SIZE, TWEET_DEFAULT_QUEUE_SIZE);
    this.tweetQueue = new LinkedBlockingQueue<Status>(queueSize);
  }

  /**
   * Start sampling tweets from Twitter. Sampling will stop (with future items being dropped)
   * as soon as the internal blocking queue is full. Use the iterator to take items out of that
   * queue.
   * */
  public void triggerSampling() {
    ConfigurationBuilder cb = new ConfigurationBuilder();
    cb.setUser(this.username);
    cb.setPassword(this.password);
    @SuppressWarnings("deprecation")
    TwitterStream twitterStream = new TwitterStreamFactory(cb.build(), this).getInstance();
    twitterStream.sample();
  }

  /**{@inheritDoc}*/
  @Override
  public Iterator<Status> iterator() {
    return this.iterator;
  }

  /**
   * Acts as connecting point between twitter4j and the twitter sink.
   * */
  private class TweetSamplingStreamIterator implements Iterator<Status> {
    /**{@inheritDoc}*/
    @Override
    public boolean hasNext() {
      int maxErrorThreshold = configuration.getIntValue(TWEET_ERROR_THRESHOLD, TWEET_DEFAULT_ERROR_THRESHOLD);

      if (failures.get() >= maxErrorThreshold) {
        LOG.error("Stopping twitter streaming. Too many errors: " + failures);
      }
      return (failures.get() < maxErrorThreshold);
    }

    /**{@inheritDoc}*/
    @Override
    public Status next() {
      int timeout = configuration.getIntValue(TWEET_TAKE_TIMEOUT, TWEET_DEFAULT_TAKE_TIMEOUT);

      try {
        Status value = tweetQueue.poll(timeout, TimeUnit.SECONDS);
        return value;
      } catch (InterruptedException e) {
        LOG.info("Failed to retrieve any tweets in " + timeout + " seconds. Increasing failure counter by one.");
        failures.incrementAndGet();
      }
      return null;
    }

    /** Operation is not supported for this iterator. */
     @Override
    public void remove() {
       throw new UnsupportedOperationException("Remove on twitter streams not possible.");
     }
  }

  /**{@inheritDoc}*/
  @Override
  public void onStatus(Status status) {
    try {
      int receiveTimeout = this.configuration.getIntValue(TWEET_RECEIVE_TIMEOUT, TWEET_DEFAULT_RECEIVE_TIMEOUT);

      if (!this.tweetQueue.offer(status, receiveTimeout, TimeUnit.SECONDS)) {
        LOG.warn("Failed to add current tweet content to queue: Dropping as queue seems full.");
      }
    } catch (InterruptedException e) {
      failures.incrementAndGet();
      LOG.warn("Could not process current tweet due to exception. Increasing failure count by one and continuing to process. " + e
          .getMessage());
    }


  }

  /**{@inheritDoc}*/
  @Override
  public void onDeletionNotice(StatusDeletionNotice statusDeletionNotice) {
    // ignoring
  }

  /**{@inheritDoc}*/
  @Override
  public void onTrackLimitationNotice(int numberOfLimitedStatuses) {
    // ignoring
  }

  /** {@inheritDoc}*/
  @Override
  public void onScrubGeo(int first, long second) {
    // ignoring
  }

  /**{@inheritDoc}*/
  @Override
  public void onException(Exception e) {
    failures.incrementAndGet();
  }
}
